package com.doit.day04

import scala.collection.immutable
import scala.io.Source

/*
数据源：
    电影数据：movies   电影id   电影名称   电影类型
    评分数据：ratings  用户id   电影id   评分多少  时间戳   维度表和事实表
    用户数据：users    用户id   性别   年龄   职业   邮编

把哪一张表作为map表呢   movies  map表

    1.求每部电影的平均分，按照降序输出  ==> 我需要得是电影名称 + 平均分

    2.每个电影类型的平均分

    3.哪些年份的电影评分(平均分)最高，取最高的前五个年份

    4.每个职业最喜欢的前三个电影id   观看次数最多的

    5.年龄段在“18-24”的男性年轻人，最喜欢看哪10部电影（Id）
 */
object _01_电影需求 {
  def main(args: Array[String]): Unit = {

    val movie_line: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\movies.txt").getLines().toList
    val rating_line: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\ratings.txt").getLines().toList
    val user_line: List[String] = Source.fromFile("C:\\Users\\34650\\Desktop\\课程资料\\资料\\users.txt").getLines().toList

    /*
       ETL
     */
    val movies: Map[String, (String, String)] = movie_line.map(line => {
      val arr: Array[String] = line.split("_")
      //电影id    电影名称   电影类型
      (arr(0), (arr(1), arr(2)))
    }).toMap

    val ratings: List[(String, String, String, String)] = rating_line.map(line => {
      val arr: Array[String] = line.split("_")
      //user_id   电影id   评分   时间戳
      (arr(0), arr(1), arr(2), arr(3))
    })


/*    xuqiu1(movies,ratings).toList
      .sortBy(_._2)
      .reverse
      .take(10)
      .foreach(println)*/

    xuqiu2(movies,ratings)


  }

  //1.求每部电影的平均分，按照降序输出
  def xuqiu1(movies: Map[String, (String, String)],ratings: List[(String, String, String, String)])={
    val grouped: Map[String, List[(String, String, String, String)]] = ratings.groupBy(_._2)
    val idAndAvg: Map[String, Double] = grouped.map(tp => {
      val movie_id: String = tp._1
      val value: List[(String, String, String, String)] = tp._2
      (movie_id, value.map(tp => tp._3.toDouble).sum / value.size)
    })

    idAndAvg.map(tp=>{
      val movie_id: String = tp._1
      val movie_name: String = movies.getOrElse(movie_id, (null, null))._1
      (movie_name,tp._2)
    })
  }



  //2.每个电影类型的平均分
  def xuqiu2(movies: Map[String, (String, String)],ratings: List[(String, String, String, String)]): Unit ={

    //一对多的关系，怎么操作呢？？如果说能够得到 电影id 和电影类型的一个一对多的关系
/*    val list: List[(String, String)] = movies.toList.flatMap(tp => {
      val movie_id: String = tp._1
      //Animation|Children's|Comedy
      val movie_type: String = tp._2._2
      val arr: Array[String] = movie_type.split("\\|")
      for (elem <- arr) yield {
        (movie_id, elem)
      }
    })*/


    //先join   100万次
    val tuples: List[(String, String)] = ratings.map(tp => {
      //根据评分表中得电影id获取到对应得电影类型
      val tuple: (String, String) = movies.getOrElse(tp._2, ("未知", "未知"))
      //电影类型和评分返回
      (tuple._2, tp._3)
    })

    //300万次
    val tuples1: List[(String, Double)] = tuples.flatMap(tp => {
      val movie_type: String = tp._1
      val score: Double = tp._2.toDouble
      val arr: Array[String] = movie_type.split("\\|")
      for (elem <- arr) yield {
        (elem, score)
      }
    })

    tuples1.groupBy(_._1)
      .map(tp=>{
        (tp._1,(tp._2.map(_._2).sum/tp._2.size).formatted("%.2f").toDouble)
      }).toList
      .sortBy(_._2)
      .reverse
      .take(10)
      .foreach(println)



    //怎么得到这个元祖呢？ 从里面拿出来==》 join
    //双重for循环  效率太他妈低啦

    //想要处理成电影类型评分这样的元祖

  }

}
